package zhaopingshuju;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.*;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MapTest extends Mapper<LongWritable, Text, Data, NullWritable> {
    private Data k = new Data();
    private Map<Integer, String> city = new HashMap();
    private int city_code;
    private int salary;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
//        URI[] uris = context.getCacheFiles();
//        FileTest file = new FileTest(uris[0]);
//        BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
        //1.获取分布式缓存列表
        URI[] cacheFiles = context.getCacheFiles();
        //2.获取分布式缓存文件的文件系统
        FileSystem fileSystem = FileSystem.get(cacheFiles[0], context.getConfiguration());
        //3.获取文件的输入流
        FSDataInputStream inputStream = fileSystem.open(new Path(cacheFiles[0]));
        //4.读取文件，并将文件写入Map集合
        //将字节字符转换为字符流
        BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));

        String line;
        while ((line = br.readLine()) != null) {
            city.put(Integer.parseInt(line.split(",")[0]), line.split(",")[1]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        JSONObject jsonObject = JSONObject.parseObject(value.toString());
        String [] datas = new String[14];
        datas[0] = jsonObject.getString("id");
        datas[1] = jsonObject.getString("company_name");
        datas[2] = jsonObject.getString("eduLevel_name");
        datas[3] = jsonObject.getString("emplType");
        datas[4] = jsonObject.getString("jobName");
        datas[5] = jsonObject.getString("salary");
        datas[6] = jsonObject.getString("createDate");
        datas[7] = jsonObject.getString("endDate");
        datas[8] = jsonObject.getString("city_code");
        datas[9] = jsonObject.getString("companySize");
        datas[10] = jsonObject.getString("welfare");
        datas[11] = jsonObject.getString("responsibility");
        datas[12] = jsonObject.getString("place");
        datas[13] = jsonObject.getString("workingExp");
        for (String s:datas){
            if (s.equals("")||s==null){
                return;
            }
        }
        k = JSON.parseObject(value.toString(),Data.class);
        city_code = k.getCity_code();
        k.setCity_name(city.get(city_code));
        String s [] = k.getSalary().split("-");
        salary = Integer.parseInt(s[1].substring(s[1].length()-2,s[1].length()-1))-Integer.parseInt(s[0].substring(s[0].length()-2,s[0].length()-1));
        k.setSalary(String.valueOf(salary));
        context.write(k,NullWritable.get());
    }
}
